package com.qing.search

import java.io.File
import java.nio.file.Paths


import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.lucene.analysis.standard.StandardAnalyzer
import org.apache.lucene.index.{DirectoryReader, IndexWriter, IndexWriterConfig, Term}
import org.apache.lucene.search.BooleanClause.Occur
import org.apache.lucene.search._
import org.apache.lucene.store._
import org.apache.spark.TaskContext

import scala.collection.mutable.ListBuffer
import scala.util.control.Breaks.{break, breakable}

/**
  * Created by wuliao on 2017/8/17.
  */
private[search] class LucenePageRDDIterator(val context: TaskContext,
                                            partition: String,
                                            params: java.util.Map[String, Array[String]],
                                            var page: Int) {
  val pageSize = 10
  var finished = false
  var writer: IndexWriter = null
  private var closed = false
  var lock: File = null

  def closeIfNeeded() {
    if (!closed) {
      close()
      closed = true
    }
  }

  protected def close() = {
    if (writer != null) {
      writer.commit()
      writer.close()
    }
  }


  val searcher = {

    val path = partition.split("&").apply(0)
    val num = partition.split("&").apply(1).toInt
    var searcher: IndexSearcher = null

    breakable {
      for (a <- 0 until num) {
        val file = new File(path + a)
        lock = new File(path + a + "/lock")
        if (file.exists() && !lock.exists()) {
          val directory = FSDirectory.open(Paths.get(path + a))
          val analyzer = new StandardAnalyzer()
          val writerConfig = new IndexWriterConfig(analyzer)
          writer = new IndexWriter(directory, writerConfig)
          val reader = DirectoryReader.open(directory)
          searcher = new IndexSearcher(reader)
          lock.createNewFile()
          break()
        }
      }
    }
    searcher
  }

  lazy val query = {
    if (params.size() == 0) {
      null
    } else {
      val booleanQueryBuilder = new BooleanQuery.Builder
      val it = params.entrySet().iterator()
      while (it.hasNext) {
        val temp = it.next
        if (temp.getKey.equals("and")) {
          for (str <- temp.getValue) {
            val array = str.split("&")
            for (i <- 0 until array.length) {
              val arr = array.apply(0).split(":")
              val query = new TermQuery(new Term(arr.apply(0), arr.apply(1)))
              booleanQueryBuilder.add(query, Occur.MUST)
            }
          }
        } else if (temp.getKey.equals("or")) {
          for (str <- temp.getValue) {
            val array = str.split("&")
            for (temp <- array) {
              val arr = temp.split(":")
              val query = new TermQuery(new Term(arr.apply(0), arr.apply(1)))
              booleanQueryBuilder.add(query, Occur.SHOULD)
            }
          }
        } else if (temp.getKey.equals("not")) {
          for (str <- temp.getValue) {
            val array = str.split("&")
            for (temp <- array) {
              val arr = temp.split(":")
              val query = new TermQuery(new Term(arr.apply(0), arr.apply(1)))
              booleanQueryBuilder.add(query, Occur.MUST_NOT)
            }
          }
        } else if (temp.getKey.equals("gt")) {
          //大于
          for (str <- temp.getValue) {
            val array = str.split("&")
            for (temp <- array) {
              val arr = temp.split(":")
              val query = TermRangeQuery.newStringRange(arr.apply(0), arr.apply(1), null, true, false);
              booleanQueryBuilder.add(query, Occur.MUST)
            }
          }
        } else if (temp.getKey.equals("lt")) {
          //小于
          for (str <- temp.getValue) {
            val array = str.split("&")
            for (temp <- array) {
              val arr = temp.split(":")
              val query = TermRangeQuery.newStringRange(arr.apply(0), null, arr.apply(1), false, true);
              booleanQueryBuilder.add(query, Occur.MUST)
            }
          }
        }
      }
      booleanQueryBuilder.build()
    }

  }


  def getLastScoreDoc(pageIndex: Int, query: Query): ScoreDoc = {
    if (pageIndex == 1) return null
    //如果是第一页就返回空
    val num = pageSize * (pageIndex - 1)
    //获取上一页的最后数量
    val tds = searcher.search(query, num)
    tds.scoreDocs.apply(num - 1)
  }


  lazy val docs = {
    if (searcher == null) {
      closeIfNeeded()
      ListBuffer[Tuple2[Float, String]]().iterator
    } else {
      if (query != null) {
        try {
          val lastSd = getLastScoreDoc(page, query)
          val tds = searcher.searchAfter(lastSd, query, pageSize)
          val list = ListBuffer[Tuple2[Float, String]]()
          for (sd <- tds.scoreDocs) {
            val doc = searcher.doc(sd.doc)
            list.append(new Tuple2[Float, String](sd.score, doc.get("_all")))
          }
          closeIfNeeded()
          list.toIterator
        } catch {
          case ex: Exception => {
            closeIfNeeded()
            ListBuffer[Tuple2[Float, String]]().iterator
          }
        }
      } else {
        val total = writer.numDocs()
        val begin = pageSize * (page - 1)
        val end = pageSize * page
        val list = ListBuffer[Tuple2[Float, String]]()
        if (begin < total) {
          if (end <= total) {
            var i = begin
            for (i <- begin until end) {
              val doc = searcher.doc(i)
              list.append(new Tuple2[Float, String](0.0f, doc.get("_all")))
            }
          } else {
            var i = begin
            for (i <- begin until total) {
              val doc = searcher.doc(i)
              list.append(new Tuple2[Float, String](0.0f, doc.get("_all")))
            }
          }
          closeIfNeeded()
          list.toIterator
        } else {
          closeIfNeeded()
          ListBuffer[Tuple2[Float, String]]().iterator
        }
      }
    }

  }

}
